package Dm

import (
	"errors"
	"fmt"
	"db2s/ETL"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/parDef"
	"db2s/ref"
	"reflect"
	"strings"
	"sync/atomic"
)

func newTimeTypeToFunc(colMap, columnName, asColumn string) (r string) {
	switch colMap {
	case "year-month-day hour:minute:second":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS') as \"%v\"", columnName, asColumn)
	case "year-month-day hour:minute:second.utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS TZR') as \"%v\"", columnName, asColumn)
	case "year-month-day hour:minute:second.milli(1)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 1, asColumn)
	case "year-month-day hour:minute:second.milli(1).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 1, asColumn)
	case "year-month-day hour:minute:second.milli(2)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 2, asColumn)
	case "year-month-day hour:minute:second.milli(2).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 2, asColumn)
	case "year-month-day hour:minute:second.milli(3)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 3, asColumn)
	case "year-month-day hour:minute:second.milli(3).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 3, asColumn)
	case "year-month-day hour:minute:second.micro(1)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 4, asColumn)
	case "year-month-day hour:minute:second.micro(1).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 4, asColumn)
	case "year-month-day hour:minute:second.micro(2)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 5, asColumn)
	case "year-month-day hour:minute:second.micro(2).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 5, asColumn)
	case "year-month-day hour:minute:second.micro(3)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 6, asColumn)
	case "year-month-day hour:minute:second.micro(3).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 6, asColumn)
	case "year-month-day hour:minute:second.nano(1)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 7, asColumn)
	case "year-month-day hour:minute:second.nano(1).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 7, asColumn)
	case "year-month-day hour:minute:second.nano(2)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 8, asColumn)
	case "year-month-day hour:minute:second.nano(2).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 8, asColumn)
	case "year-month-day hour:minute:second.nano(3)":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d') as \"%v\"", columnName, 9, asColumn)
	case "year-month-day hour:minute:second.nano(3).utc":
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR') as \"%v\"", columnName, 9, asColumn)
	default:
		r = fmt.Sprintf("to_char(%s,'YYYY-MM-DD HH24:MI:SS') as \"%v\"", columnName, asColumn)
	}
	return
}
func newTimeValueToFunc(colMap, value string) (r string) {
	switch colMap {
	case "year-month-day hour:minute:second":
		r = fmt.Sprintf("to_date(%s,'YYYY-MM-DD HH24:MI:SS')", value)
	case "year-month-day hour:minute:second.utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS TZR')", value)
	case "year-month-day hour:minute:second.milli(1)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 1)
	case "year-month-day hour:minute:second.milli(1).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 1)
	case "year-month-day hour:minute:second.milli(2)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 2)
	case "year-month-day hour:minute:second.milli(2).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 2)
	case "year-month-day hour:minute:second.milli(3)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 3)
	case "year-month-day hour:minute:second.milli(3).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 3)
	case "year-month-day hour:minute:second.micro(1)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 4)
	case "year-month-day hour:minute:second.micro(1).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 4)
	case "year-month-day hour:minute:second.micro(2)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 5)
	case "year-month-day hour:minute:second.micro(2).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 5)
	case "year-month-day hour:minute:second.micro(3)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 6)
	case "year-month-day hour:minute:second.micro(3).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 6)
	case "year-month-day hour:minute:second.nano(1)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 7)
	case "year-month-day hour:minute:second.nano(1).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 7)
	case "year-month-day hour:minute:second.nano(2)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 8)
	case "year-month-day hour:minute:second.nano(2).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 8)
	case "year-month-day hour:minute:second.nano(3)":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d')", value, 9)
	case "year-month-day hour:minute:second.nano(3).utc":
		r = fmt.Sprintf("to_timestamp(%s,'YYYY-MM-DD HH24:MI:SS.FF%d TZR')", value, 9)
	default:
		r = fmt.Sprintf("to_date(%s,'YYYY-MM-DD HH24:MI:SS')", value)
	}
	return
}
func newParameterFmToString(n int64, p string) string {
	var s []string
	if n > 38 {
		n = 38
	}
	for i := int64(0); i < n; i++ {
		s = append(s, p)
	}
	if len(s) == 0 {
		return ""
	}
	return fmt.Sprintf("%v", strings.Join(s, ""))
}
func newScaleFmToString(n int64, p string) string {
	var s []string
	if n > 38 {
		n = 38
	}
	for i := int64(0); i < n; i++ {
		s = append(s, p)
	}
	if len(s) == 0 {
		return ""
	}
	return fmt.Sprintf(".%v", strings.Join(s, ""))
}
func newDecimalTypeToFunc(precision, scale int64, colMap, columnName, asColumn string) (r string) {
	switch colMap {
	case "decimal(precision,scale)":
		r = fmt.Sprintf("to_char(%s,'FM%s0%s') as \"%v\" ", columnName, newParameterFmToString(precision-scale, "9"), newScaleFmToString(scale, "0"), asColumn)
	case "decimal(max,max)":
		r = fmt.Sprintf("to_char(%s) as \"%v\" ", columnName, asColumn)
	default:
		r = fmt.Sprintf("%v as \"%v\"", columnName, asColumn)
	}
	return
}

func NewSpecialTypeToFunc(LockCol parDef.Encryption, colMap parDef.ColMetaMapS) (r string) {
	asColumn := fmt.Sprintf("C%v", colMap.ColumnSeq)
	var unlockFunc string
	for _, v := range LockCol.Column {
		if strings.EqualFold(v, colMap.ColumnName) && len(LockCol.UnlockFunc) > 0 {
			unlockFunc = fmt.Sprintf("%v.%v(\"%v\")", LockCol.Schema, LockCol.UnlockFunc, colMap.ColumnName)
			break
		}
	}
	if len(unlockFunc) == 0 {
		unlockFunc = fmt.Sprintf("\"%v\"", colMap.ColumnName)
	}
	switch colMap.TypeBelong {
	case "time", "timestamp":
		return newTimeTypeToFunc(colMap.Type.ColTypeMap, unlockFunc, asColumn)
	case "decimal":
		return newDecimalTypeToFunc(colMap.Type.Precision, colMap.Type.Scale, colMap.Type.ColTypeMap, unlockFunc, asColumn)
	case "char":
		return fmt.Sprintf("rtrim(%s) as \"%v\"", unlockFunc, asColumn)
	default:
		return fmt.Sprintf("%v as \"%v\"", unlockFunc, asColumn)
	}
}

// NewSpecialCharValueToFunc 根据列的数据类型，将字符串值继续函数转换
func NewSpecialCharValueToFunc(colMap parDef.ColMetaMapS, value string) (newValue string) {
	newValue = fmt.Sprintf("'%v'", strings.ReplaceAll(value, "'", "\\'"))
	switch colMap.TypeBelong {
	case "time", "timestamp":
		return newTimeValueToFunc(colMap.Type.ColTypeMap, newValue)
	default:
		switch {
		case strings.EqualFold(colMap.ColumnName, "rowId"):
			newValue = value
		case strings.EqualFold(newValue, "NULL"):
			newValue = value
		}
		return
	}
}
func newHitSpecify(indexName, partitionName string) (hitIndexName string) {
	//指定partition name会导致无法使用索引，进行全表扫描，且该sql为查询数据，已经通过索引列范围确定查询范围，所以不需要指定分区
	if len(indexName) > 0 && !strings.EqualFold(partitionName, "single") && len(partitionName) > 0 {
		hitIndexName = fmt.Sprintf(" /*+ index(T %v) */ ", indexName)
	}
	return
}

func newLeftStitching(s parDef.WhereGenerateInput) (l string) {
	var whereLeft []string
	for _, v := range func() []map[string]string {
		if len(s.NewLeftDirection) > 0 {
			return s.NewLeftDirection
		}
		return getAtomicValue(s.LeftDirection)
	}() {
		columnName, nLeftOk := v["columnName"]
		columnDataLeft, cLeftOk := v["columnDataLeft"]
		querySymbolLeft, qLeftOk := v["querySymbolLeft"]
		if cLeftOk && qLeftOk && nLeftOk {
			if strings.EqualFold(querySymbolLeft, "=") && strings.EqualFold(columnDataLeft, "NULL") { //处理等值null
				whereLeft = append(whereLeft, fmt.Sprintf("\"%s\" %s %s", columnName, " IS ", columnDataLeft))
			} else if strings.EqualFold(querySymbolLeft, ">") && strings.EqualFold(columnDataLeft, "NULL") {
				whereLeft = append(whereLeft, fmt.Sprintf("\"%s\" %s %s", columnName, " IS NOT ", columnDataLeft))
			} else {
				whereLeft = append(whereLeft, fmt.Sprintf("\"%s\" %s %s", columnName, querySymbolLeft, NewSpecialCharValueToFunc(getSpecialColNameMetaMap(columnName, s.ColMetaMap), columnDataLeft)))
			}
		}
	}
	return strings.Join(whereLeft, " and ")
}
func getSpecialColNameMetaMap(name string, colMeta []parDef.ColMetaMapS) (r parDef.ColMetaMapS) {
	for _, col := range colMeta {
		if strings.EqualFold(col.ColumnName, name) {
			return col
		}
	}
	return
}
func newRightStitching(s parDef.WhereGenerateInput) string {
	var whereRight []string
	for _, v := range func() []map[string]string {
		if len(s.NewRightDirection) > 0 {
			return s.NewRightDirection
		}
		return getAtomicValue(s.RightDirection)
	}() {
		columnName, nRightOk := v["columnName"]
		columnDataRight, cRightOk := v["columnDataRight"]
		querySymbolRight, qRightOk := v["querySymbolRight"]
		if cRightOk && qRightOk && nRightOk {
			if strings.EqualFold(querySymbolRight, "=") && strings.EqualFold(columnDataRight, "NULL") {
				whereRight = append(whereRight, fmt.Sprintf("\"%s\" %s %s", columnName, " IS ", columnDataRight))
			} else if strings.EqualFold(querySymbolRight, ">") && strings.EqualFold(columnDataRight, "NULL") {
				whereRight = append(whereRight, fmt.Sprintf("\"%s\" %s %s", columnName, " IS NOT ", columnDataRight))
			} else {
				whereRight = append(whereRight, fmt.Sprintf("\"%s\" %s %s", columnName, querySymbolRight, NewSpecialCharValueToFunc(getSpecialColNameMetaMap(columnName, s.ColMetaMap), columnDataRight)))
			}
		}
	}
	return strings.Join(whereRight, " and ")
}
func newWhereGenerate(whereSql string, s parDef.WhereGenerateInput) (ConditionSlice []string, err error) {
	if len(whereSql) > 0 {
		ConditionSlice = append(ConditionSlice, whereSql)
		return
	}
	if v := newLeftStitching(s); len(v) > 0 {
		ConditionSlice = append(ConditionSlice, v)
	}
	if v := newRightStitching(s); len(v) > 0 {
		ConditionSlice = append(ConditionSlice, v)
	}
	return
}
func newOrderBY(s parDef.WhereGenerateInput) (res string, err error) {
	var orderByColumn []string
	if strings.EqualFold(s.NullConstraint, "null") {
		return
	}
	var uniqOrderByColumn = make(map[string]int)
	for _, v := range s.IndexColumn {
		for _, v1 := range getAtomicValue(s.LeftDirection) {
			if v2, ok := v1["columnName"]; ok {
				if strings.EqualFold(v, v2) {
					if _, ok2 := uniqOrderByColumn[v]; !ok2 {
						uniqOrderByColumn[v]++
						orderByColumn = append(orderByColumn, v)
					}
				}
			}
		}
	}
	if len(orderByColumn) > 0 {
		return fmt.Sprintf("order by \"%v\"", strings.Join(orderByColumn, "\",\"")), nil
	}
	return
}
func newResultCol(asSelectColumn []string, tableAS string) (r string) {
	var l []string
	for _, v := range asSelectColumn {
		l = append(l, fmt.Sprintf("%v.\"%v\" ", tableAS, v))
	}
	return strings.Join(l, ",")
}
func newSelectCol(selectC []string) (r string) {
	var selectIndexColumn []string
	for _, v := range selectC {
		selectIndexColumn = append(selectIndexColumn, fmt.Sprintf("\"%v\"", v))
	}
	return strings.Join(selectIndexColumn, ",")
}
func newOrderByNull(nullBool *int32, indexCol []string) (r string) {
	var nullOrderDesc []string
	for k, v := range indexCol {
		if k > 0 && atomic.LoadInt32(nullBool) == 1 {
			nullOrderDesc = append(nullOrderDesc, fmt.Sprintf("\"%v\" ASC NULLS FIRST", v))
		} else {
			nullOrderDesc = append(nullOrderDesc, fmt.Sprintf("\"%v\"", v))
		}
	}
	return fmt.Sprintf(" order by %v", strings.Join(nullOrderDesc, ","))
}
func newWhereSplicing(whereCondition []string) (r string) {
	if len(whereCondition) > 0 {
		return fmt.Sprintf(" where %v", strings.Join(whereCondition, " and "))
	}
	return
}
func getIndexColMetaCharset(parameter parDef.Parameter) (collation string, compareVal []string) {
	switch parameter.Meta.(type) {
	case map[string][]string:
		for col, val := range parameter.Meta.(map[string][]string) {
			for _, subCol := range parameter.Object.Index.ColumnMate {
				if strings.EqualFold(col, subCol.ColumnName) {
					collation = subCol.Collation
					compareVal = val
					break
				}
			}
		}
	}
	return
}
func getColCharacterVarchar(parameter parDef.Parameter) (newCompareVal []string) {
	var collation, compareVal = getIndexColMetaCharset(parameter)
	for _, v := range compareVal {
		var notNullVal = v
		if strings.EqualFold(v, "<null>") {
			notNullVal = "NULL"
		}
		newCompareVal = append(newCompareVal, fmt.Sprintf("NLSSORT('%v', 'NLS_SORT = %v')", notNullVal, collation))
	}
	return
}
func (rows RowsF) Prefix(parameter parDef.Parameter) (result global.Return, err error) {
	var (
		tableName, ignoreStr string
		insertColumn         []string
	)
	if result, err = rows.InsertAllCol(parameter); err != nil {
		err = ref.ErrAddPrintf("DM.Prefix", err)
		return
	}
	switch result.Result.(type) {
	case []string:
		insertColumn = result.Result.([]string)
	default:
		err = ref.ErrAddPrintf("DM.Prefix", errors.New(fmt.Sprintf("type not match. curry type is %v", reflect.TypeOf(result.Result))))
	}
	tableName = fmt.Sprintf(" \"%v\".\"%v\" ", parameter.Object.Schema, parameter.Object.Table)
	result.Result = fmt.Sprintf("%s ", fmt.Sprintf("INSERT %v INTO %v (\"%s\") VALUES", ignoreStr, tableName, strings.Join(insertColumn, "\",\"")))
	if parameter.Object.LobLogo {
		var s1 []string
		for i := 0; i < len(insertColumn); i++ {
			s1 = append(s1, "?")
		}
		result.Result = fmt.Sprintf("INSERT INTO %v (\"%s\") VALUES (%s)", tableName, strings.Join(insertColumn, "\",\""), strings.Join(s1, ","))
	}
	return
}

func (rows RowsF) writeFileDML(parameter parDef.Parameter) (result global.Return, err error) {
	var event = "[writeFileDML]"
	if parameter.ExecDML.FilePoint == nil {
		return
	}
	if err = parameter.ExecDML.FilePoint.Write("", parameter.ExecDML.Sql); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return
}

func (rows RowsF) connExecDML(parameter parDef.Parameter) (result global.Return, err error) {
	//var event = "[connExecDML]"
	//if result.TimeOut, result.Result, err = newExecUpdateSql(ExecSqlInputS{
	//	Ss:           parameter,
	//	Event:        event,
	//	UpdateSqlStr: []string{parameter.ExecDML.Sql},
	//}); err != nil {
	//	err = ref.ErrAddPrintf(event, err)
	//	return
	//}
	return
}

func (rows RowsF) ExecDML(parameter parDef.Parameter) (result global.Return, err error) {
	var event = "[ExecDML]"
	switch parameter.Options.RepairMethod {
	case "file":
		return rows.writeFileDML(parameter)
	case "table":
		return rows.connExecDML(parameter)
	default:
		err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("The repair method was not matched. curry method is %v", parameter.Options.RepairMethod)))
	}
	return
}
func (rows RowsF) SelectIdxCol(parameter parDef.Parameter) (result global.Return, err error) {
	var columnName, asColumnName []string
	for _, i := range parameter.Object.IndexCol {
		p := getSpecialColNameMetaMap(i, parameter.Object.Column)
		asColumnName = append(asColumnName, fmt.Sprintf("C%v", p.ColumnSeq))
		columnName = append(columnName, NewSpecialTypeToFunc(parameter.Object.LockCol, p))
	}
	result.Result = [][]string{columnName, asColumnName}
	return
}
func (rows RowsF) SelectAllCol(parameter parDef.Parameter) (result global.Return, err error) {
	var columnName, asColumnName []string
	for _, v := range parameter.Object.Column {
		asColumnName = append(asColumnName, fmt.Sprintf("C%v", v.ColumnSeq))
		columnName = append(columnName, NewSpecialTypeToFunc(parameter.Object.LockCol, v))
	}
	result.Result = columnName
	return
}
func (rows RowsF) InsertAllCol(parameter parDef.Parameter) (result global.Return, err error) {
	var columnName []string
	switch parameter.Meta.(type) {
	case []parDef.ColMetaMapS:
		for _, v := range parameter.Meta.([]parDef.ColMetaMapS) {
			columnName = append(columnName, v.ColumnName)
		}
	}
	result.Result = columnName
	return
}
func (rows RowsF) StringValueCompare(parameter parDef.Parameter) (result global.Return, err error) {
	var event = "[StringValueCompare]"
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	var newCompareVal = getColCharacterVarchar(parameter)
	var execSql = []string{fmt.Sprintf("SELECT \n    "+
		"CASE \n        "+
		"WHEN %v < %v "+
		"THEN '-1'\n        "+
		"WHEN %v > %v "+
		"THEN '1'\n        "+
		"ELSE '0'\n    "+
		"END AS comparison_result\n"+
		"FROM dual",
		newCompareVal[0], newCompareVal[1],
		newCompareVal[0], newCompareVal[1])}
	if result, err = forExecQuerySql(event, parameter, ETL.String, execSql); err != nil {
		err = ref.ErrAddPrintf("DM.StringValueCompare", err)
		return
	}
	if result.Result, err = stringAnyToInt64(result.Result); err != nil {
		err = ref.ErrAddPrintf("DM.StringValueCompare", err)
		return
	}
	return
}
func (rows RowsF) IndexPart(parameter parDef.Parameter) (result global.Return, err error) {
	var (
		event          = "[IndexPart]"
		res            global.Return
		selectCol      []string
		selectColAs    []string
		whereCondition []string
		indexP         parDef.IndexPart
	)
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	switch parameter.Meta.(type) {
	case parDef.IndexPart:
		indexP = parameter.Meta.(parDef.IndexPart)
	default:
		err = ref.ErrAddPrintf("Oracle.IndexPart", errors.New(fmt.Sprintf("type not matched. curry type is %v", reflect.TypeOf(parameter.Meta))))
		return
	}
	LimitSeq1 := atomic.LoadInt64(indexP.Limit.Pagination) + 1
	if res, err = rows.SelectIdxCol(parameter); err != nil {
		err = ref.ErrAddPrintf("DM.IndexPart", err)
		return
	} else if res.Result == nil {
		return
	} else {
		selectCol, selectColAs = res.Result.([][]string)[0], res.Result.([][]string)[1]
	}
	if whereCondition, err = newWhereGenerate(parameter.Options.WhereSql, indexP.WhereGenerate); err != nil {
		err = ref.ErrAddPrintf("DM.IndexPart", err)
		return
	}
	if len(whereCondition) == 0 {
		whereCondition = append(whereCondition, fmt.Sprintf("\"%s\" IS NOT NULL ", parameter.Object.IndexCol[0]))
	}
	var execSql []string
	execSql = append(execSql, fmt.Sprintf("select * from ( \n"+
		"select %v,rownum rn from ( \n"+
		"select %v %s from ( \n"+
		"select %v from %v %v where 1 = 1 %v) T %s %s) t \n"+
		"where rownum<=%d) \n"+
		"where rn>%d",
		newResultCol(selectColAs, "t"), newHitSpecify(indexP.Index.IndexName, parameter.Object.Partition), strings.Join(selectCol, ","),
		newSelectCol(parameter.Object.IndexCol), tableObject(parameter.Object.Schema, parameter.Object.Table, parameter.Object.Partition), specifyScn(parameter.Options.Scn), rowsLimitC(parameter.Options.RowsLimit),
		newWhereSplicing(whereCondition), newOrderByNull(indexP.WhereGenerate.NullBool, parameter.Object.IndexCol),
		LimitSeq1,
		atomic.LoadInt64(indexP.Limit.Pagination)))
	if result, err = forExecQuerySql(event, parameter, ETL.Map, execSql); err != nil {
		err = ref.ErrAddPrintf("DM.IndexPart", err)
	}
	return
}

func (rows RowsF) SelectRowsSql(parameter parDef.Parameter) (result global.Return, err error) {
	var (
		res                       global.Return
		event                     = "[SelectRowsSql]"
		w                         = []string{" "}
		selectCol, whereCondition []string
		orderByCol                string
		s                         parDef.IndexPart
		indexP                    parDef.IndexPart
	)
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	switch parameter.Meta.(type) {
	case parDef.IndexPart:
		indexP = parameter.Meta.(parDef.IndexPart)
	default:
		err = ref.ErrAddPrintf("DM.SelectRowsSql", errors.New(fmt.Sprintf("type not matched. curry type is %v", reflect.TypeOf(parameter.Meta))))
		return
	}
	if res, err = rows.SelectAllCol(parameter); err != nil {
		err = ref.ErrAddPrintf("DM.SelectRowsSql", err)
		return
	} else if res.Result == nil {
		return
	} else {
		selectCol = res.Result.([]string)
	}
	whereCondition, err = newWhereGenerate(parameter.Options.WhereSql, indexP.WhereGenerate)
	orderByCol, err = newOrderBY(s.WhereGenerate)
	result.Result = fmt.Sprintf("select %v %v \n"+
		"from %v %v \n"+
		"where 1 = 1 %v \n"+
		"%v",
		newHitSpecify("", parameter.Object.Partition), strings.Join(selectCol, ","),
		tableObject(parameter.Object.Schema, parameter.Object.Table, parameter.Object.Partition), specifyScn(parameter.Options.Scn),
		strings.Join(append(w, whereCondition...), " and "),
		orderByCol)
	return
}
func (rows RowsF) ReadData(parameter parDef.Parameter) (result global.Return, err error) {
	var event = "[ReadData]"
	var execSql []string
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	if parameter.Meta == nil {
		return
	}
	switch parameter.Meta.(type) {
	case string:
		execSql = append(execSql, parameter.Meta.(string))
	case []string:
		execSql = parameter.Meta.([]string)
	default:
		err = ref.ErrAddPrintf("DM.ReadData", errors.New(fmt.Sprintf("type not matched. curry type %v", reflect.TypeOf(parameter.Meta))))
		return
	}
	if result, err = forExecQuerySql(event, parameter, ETL.GarthStringPointGarth, execSql); err != nil {
		err = ref.ErrAddPrintf("DM.ReadData", err)
		return
	}
	return
}
func (rows RowsF) ReadMiss(parameter parDef.Parameter) (result global.Return, err error) {
	var (
		event   = "[DM.ReadMiss]"
		res     global.Return
		execSql []string
	)
	defer func() {
		if r := recover(); r != nil {
			err = errors.New(fmt.Sprintf("%v defer recover An exception was captured. abnormal is %v ", event, r))
		}
	}()
	if res, err = rows.SelectAllCol(parameter); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	if res.Result == nil {
		log.MainLog().Warn(event, "no data to read")
		return
	}
	execSql = append(execSql, fmt.Sprintf("select %v %v \n"+
		"from %v %v \n"+
		"where 1= 1 %v",
		parallel(parameter.Options.TablePx), strings.Join(res.Result.([]string), ","),
		tableObject(parameter.Object.Schema, parameter.Object.Table, parameter.Object.Partition), specifyScn(parameter.Options.Scn),
		rowsLimitC(parameter.Options.RowsLimit)))
	log.MainLog().Debug(event, fmt.Sprintf(" sql is %v", execSql))
	if result, err = forExecQuerySql(event, parameter, ETL.FlowAny, execSql); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return
}
func (rows RowsF) InsertSql(parameter parDef.Parameter) (result global.Return, err error) {
	return
}
func (rows RowsF) Insert(parameter parDef.Parameter) (result global.Return, err error) {
	var (
		setVariables []string
		event        = "[ApplyInsert]"
		execSql      []*string
	)
	for _, v := range setVariables {
		execSql = append(execSql, &v)
	}
	switch parameter.Meta.(type) {
	case []*string:
		for _, v := range parameter.Meta.([]*string) {
			execSql = append(execSql, v)
		}
	default:
		err = ref.ErrAddPrintf("DM.Insert", errors.New(fmt.Sprintf("type not match. type is %v", reflect.TypeOf(parameter.Meta))))
		return
	}
	if result, err = newExecInsertSql(ExecSqlInputS{
		Ss:           parameter,
		Event:        event,
		InsertSqlStr: execSql,
	}); err != nil {
		err = ref.ErrAddPrintf("DM.Insert", err)
		return
	}
	return
}
